package com.cn.client;

import com.cn.client.impl.DefaultMQPushConsumerImpl;
import com.cn.common.MessageExt;
import com.cn.common.MessageQueue;
import com.cn.common.ProcessQueue;
import com.cn.common.PullRequest;

import java.util.List;
import java.util.concurrent.*;

public class ConsumeMessageService {


    private final ThreadPoolExecutor threadPoolExecutor;
    private final BlockingQueue<Runnable> queue;
    private final MQClientInstance mqClientInstance;
    private final DefaultMQPushConsumerImpl defaultMQPushConsumer;
    private final MessageQueueLock messageQueueLock = new MessageQueueLock();
    private final ScheduledExecutorService scheduledExecutorService;

    public ConsumeMessageService(MQClientInstance mqClientInstance, DefaultMQPushConsumerImpl defaultMQPushConsumer) {
        this.mqClientInstance = mqClientInstance;
        this.defaultMQPushConsumer = defaultMQPushConsumer;
        this.queue = new LinkedBlockingQueue<>();
        this.threadPoolExecutor = new ThreadPoolExecutor(4, 8, 1, TimeUnit.MINUTES, queue);
        this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
    }

    public void submit(PullRequest pullRequest) {
        ConsumeProcess consumeProcess = new ConsumeProcess(pullRequest.getMessageQueue(), pullRequest.getProcessQueue());
        this.threadPoolExecutor.submit(consumeProcess);
    }

    public void start() {
        //定时锁定队列，到broker端给队列加锁
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                //调用接口，到broker端，对当前客户端持有的接口进行锁定，确保可用
                defaultMQPushConsumer.getRemplanceImpl().lockAll();
            }
        }, 10, 20, TimeUnit.SECONDS);//延迟5秒，每20秒执行一次
    }

    public void processConsumeResult(List<MessageExt> msgs, MessageConsumeResultStatus status, ConsumeProcess process) {
        Long offset = -1l;
        switch (status) {
            case commit:
                //成功了
                offset = process.processQueue.commit();
                break;
            case rollback:
                //回滚，重新消费
                process.processQueue.rollback(msgs);
                break;
            case error:

                break;
        }
        if (offset > 0) {
            //保存本次消费位移到
            defaultMQPushConsumer.getOffsetStore().updateOffset(process.getMessageQueue(), offset);
        }
    }


    class ConsumeProcess implements Runnable {
        private MessageQueue messageQueue;
        private ProcessQueue processQueue;

        public MessageQueue getMessageQueue() {
            return messageQueue;
        }

        public ProcessQueue getProcessQueue() {
            return processQueue;
        }

        public ConsumeProcess(MessageQueue messageQueue, ProcessQueue processQueue) {
            this.messageQueue = messageQueue;
            this.processQueue = processQueue;
        }

        @Override
        public void run() {
            //确保该队列 当前消费只能被一个线程执行
            Object obj = messageQueueLock.lock(this.messageQueue);
            synchronized (obj) {
                for (boolean flag = true; flag; ) {
                    if (!processQueue.isLocked()) {
                        //服务已经停止了
                        System.out.println("1/进程已经失效了");
                        return;
                    }
                    if (processQueue.isDropped()) {
                        System.out.println("2/进程已经失效了");
                        return;
                    }

                    //每次消费一条
                    List<MessageExt> msgs = processQueue.takeMessage(1);
                    if (!msgs.isEmpty()) {
                        //获取锁
                        this.processQueue.getLockConsume().lock();
                        MessageConsumeResultStatus status = null;
                        try {
                            if (this.processQueue.isDropped()) {
                                break;
                            }
                            status = defaultMQPushConsumer.getMessageConsumeService().execute(msgs);
                        } finally {
                            this.processQueue.getLockConsume().unlock();
                        }
                        processConsumeResult(msgs, status, this);
                    } else {
                        flag = false;
                    }

                }
            }

        }
    }
}
